package com.zshield.config;

import com.zshield.run.KafkaPreCompute;
import com.zshield.util.Etime;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.time.LocalDate;
import java.util.Properties;

public class KafkaConfig {
    public static final String es_type = "datamap";
    public static final String es_tmpl = "datamap_precompute*";

    public static final int BULK_SIZE = 3000;
    public static final int BULK_INTERVAL = 60 * 1000;

    public static String getIndex() {
        LocalDate nowDate = LocalDate.now();
        return "datamap_precompute" + "-" + nowDate.format(Etime.DATE_FORMATTER_INDEX);
    }

    public static String getTreeIndex() {
        LocalDate nowDate = LocalDate.now();
        return "datamap_precompute_tree" + "-" + nowDate.format(Etime.DATE_FORMATTER_INDEX);
    }

    public static Properties getProducerConfig() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,KafkaPreCompute.bootstrap_server);
        props.put(ProducerConfig.CLIENT_ID_CONFIG,"sensor-recompute");
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,30000);
        props.put(ProducerConfig.ACKS_CONFIG,"all");
        props.put(ProducerConfig.RETRIES_CONFIG,Integer.MAX_VALUE);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        return  props;
    }

    public static Properties getConsumerConfig() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaPreCompute.bootstrap_server);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "sensor-es");
        props.put(ConsumerConfig.CLIENT_ID_CONFIG,KafkaPreCompute.consumer_client_id);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,300000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        return props;
    }

    public static Properties getStreamConfig() {
        Properties props = new Properties();
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaPreCompute.bootstrap_server);
        props.put(StreamsConfig.APPLICATION_ID_CONFIG,"sensor-stream");
        props.put(StreamsConfig.CLIENT_ID_CONFIG,KafkaPreCompute.stream_client_id);
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,2);
        props.put(StreamsConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
        props.put(StreamsConfig.consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG),300000);
        props.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG),300000);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
        return  props;
    }

    public static void printErrorLog(Logger logger,Throwable e) {
        StringWriter sw = null;
        PrintWriter pw = null;
        try {
            sw = new StringWriter();
            pw =  new PrintWriter(sw);
            e.printStackTrace(pw);
            pw.flush();
            sw.flush();
        } finally {
            if (sw != null) {
                try {
                    sw.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
            if (pw != null) {
                pw.close();
            }
        }
        logger.error(sw.toString());
    }
}
